package com.oscar

import org.apache.spark.graphx.{Edge, Graph, VertexId}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object SecondGraphX {

  def main(args: Array[String]): Unit = {

    //设置运行环境
    val conf = new SparkConf().setAppName("FirstGraphX").setMaster("local")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

    //设置users顶点
    val users: RDD[(VertexId, (String, Int))] =
      sc.parallelize(Array((1L, ("Alice", 28)), (2L, ("Bob", 27)), (3L, ("Charlie", 65)),
                           (4L, ("David", 42)), (5L, ("Ed", 55)), (6L, ("Fran", 50))))

    //设置relationships边
    val relationships: RDD[Edge[Int]] =
      sc.parallelize(Array(Edge(2L, 1L, 7), Edge(2L, 4L, 2), Edge(3L, 2L, 4), Edge(3L, 6L, 3),
                           Edge(4L, 1L, 1),Edge(5L, 2L, 2),Edge(5L, 3L, 8),Edge(5L, 6L, 3)))


    // 定义默认的作者,以防与不存在的作者有relationship边
    val defaultUser = ("John Doe", 0)

    // Build the initial Graph
    val graph: Graph[(String, Int), Int] = Graph(users, relationships,defaultUser)

    println("---------------------------------------------")

    println("使用mapEdges函数遍历所有的边，新增加一个属性值然后构建出新的图")

      val graph2: Graph[(String, Int), (Int, Boolean)] = graph.mapEdges(e => (e.attr *2, false))

    graph2.edges.collect.foreach(println(_))

    println("---------------------------------------------")

    println("使用mapTriplets函数遍历所有的三元组，新增加一个属性值，然后返回新的图")

    var graph3: Graph[(String, Int), (String, Boolean)]
    = graph.mapTriplets(triplet => (triplet.attr + "eee",true))

//    var value: Graph[(String, Int), String] =
//      graph.mapTriplets(triplet => (triplet.attr + "   " +
//                                    triplet.srcAttr._1 + "   " +
//                                    triplet.srcAttr._2 + "   " +
//                                    triplet.dstAttr._1 + "   " +
//                                    triplet.dstAttr._2))
    var value: Graph[(String, Int), String] =
      graph.mapTriplets(triplet => (triplet.attr*2 + "   " +
                                    triplet.srcAttr._1 + "   " +
                                    triplet.srcAttr._2*2 + "   " +
                                    triplet.dstAttr._1 + "   " +
                                    triplet.dstAttr._2*3))
    
    graph3.edges.collect.foreach(println(_))
    println("---------------------------------------------")
    value.edges.collect.foreach(println(_))
    //println("****************************************")
    //var tuples: Array[(VertexId, Int)] = graph.outDegrees.collect()
    //tuples.foreach(println(_))

    //graph.edges.collect.foreach(println(_))
    //println("****************************************")



  }

}